package com.minelx.landisk.filesystem.core.provider;

import com.minelx.landisk.filesystem.core.FileAccessingException;
import com.minelx.landisk.filesystem.core.values.PathDescription;
import org.apache.hadoop.fs.*;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.StandardOpenOption;
import java.util.*;

class HDFSPathMethodProvider implements PathMethodProvider {
	private final FileSystem fileSystem;

	private final String schemaAndAuthority;

	private final PathDescription description;

	HDFSPathMethodProvider(FileSystem fileSystem, String schemaAndAuthority, PathDescription description) {
		this.fileSystem = fileSystem;
		this.schemaAndAuthority = schemaAndAuthority;
		this.description = description;
	}

	@Override
	public String uri() {
		return schemaAndAuthority + description.toText();
	}

	@Override
	public OutputStream openOutputStream(StandardOpenOption openOption) {
		if (openOption != StandardOpenOption.CREATE) {
			throw new UnsupportedOperationException("can't open hdfs file with: " + openOption);
		}
		// returns hdfs output stream
		try {
			return fileSystem.create(getHdfsPath());
		} catch (IOException e) {
			throw new FileAccessingException("error while obtaining output stream.", e);
		}
	}

	@Override
	public InputStream openInputStream() {
		try {
			return fileSystem.open(getHdfsPath());
		} catch (IOException e) {
			throw new FileAccessingException("error while obtaining input stream.", e);
		}
	}

	@Override
	public void createFile() {
		try {
			fileSystem.create(getHdfsPath());
		} catch (IOException e) {
			throw new FileAccessingException("error while creating file.", e);
		}
	}

	@Override
	public void deleteFile() {
		try {
			fileSystem.delete(getHdfsPath(), false);
		} catch (IOException e) {
			throw new FileAccessingException("error while deleting file at: " + getHdfsPath(), e);
		}
	}

	@Override
	public void createDirectory() {
		deleteDirectory();

		// create dir after delete
		try {
			fileSystem.mkdirs(getHdfsPath());
		} catch (IOException e) {
			throw new FileAccessingException("error while creating dir.", e);
		}
	}

	@Override
	public boolean exists() {
		try {
			return fileSystem.exists(getHdfsPath());
		} catch (IOException e) {
			throw new FileAccessingException("error while verifying path exists or not.", e);
		}
	}

	@Override
	public void deleteDirectory() {
		try {
			// delete dir recursively
			fileSystem.delete(getHdfsPath(), true);
		} catch (IOException e) {
			throw new FileAccessingException("error while deleting dir recursively.", e);
		}
	}

	@Override
	public boolean isFile() {
		return getFileStatus().isFile();
	}

	private FileStatus getFileStatus() {
		try {
			return fileSystem.getFileStatus(getHdfsPath());
		} catch (IOException e) {
			throw new FileAccessingException("error while fetching status of path: " + getHdfsPath(), e);
		}
	}

	@Override
	public boolean isDirectory() {
		return getFileStatus().isDirectory();
	}

	@Override
	public String[] listDir() {
		try {
			RemoteIterator<LocatedFileStatus> result = fileSystem.listFiles(getHdfsPath(), false);
			List<String> re = new LinkedList<>();
			while (result.hasNext()) {
				LocatedFileStatus fileStatus = result.next();
				re.add(fileStatus.getPath().getName());
			}
			return re.toArray(new String[0]);
		} catch (IOException e) {
			throw new FileAccessingException("error while listing dir.", e);
		}
	}

	@Override
	public void rename(String name) {
		try {
			fileSystem.rename(getHdfsPath(), new Path(description.sibling(name).toText()));
		} catch (IOException e) {
			throw new FileAccessingException("error while renaming file.", e);
		}
	}

	private Path getHdfsPath() {
		return new Path(description.toText());
	}
}
